package com.atguigu.app.dwd.db;

import com.atguigu.utils.KafkaUtil;
import com.atguigu.utils.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneId;

/**
 * ClassName: TradePayDetailSuc
 * Package: com.atguigu.app.dwd
 * Description:
 *
 * @Author Lovxy
 * @Create 2023/5/16 19:36
 * @Version 1.0
 */
public class DwdTradePayDetailSuc {
    public static void main(String[] args) throws Exception {

        // TODO 1. 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment
                .getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.getConfig().setLocalTimeZone(ZoneId.of("GMT+8"));

        // TODO 2. 状态后端设置
//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                3, Time.days(1), Time.minutes(1)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage(
//                "hdfs://hadoop102:8020/ck"
//        );
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        // TODO 3. 读取 Kafka dwd_trade_order_detail 主题数据，封装为 Flink SQL 表
        tableEnv.executeSql("" +
                "create table dwd_trade_order_detail(\n" +
                " id STRING,\n" +
                " course_id STRING,\n" +
                " course_name STRING,\n" +
                " order_id STRING,\n" +
                " user_id STRING,\n" +
                " origin_amount STRING,\n" +
                " coupon_reduce STRING,\n" +
                " final_amount STRING,\n" +
                " session_id STRING,\n" +
                " create_time STRING,\n" +
                " update_time STRING,\n" +
                " privince_id STRING,\n" +
                " order_status STRING,\n" +
                " ts string,\n" +
                " sc STRING,\n" +
                " row_time as TO_TIMESTAMP(FROM_UNIXTIME(cast(ts as bigint)*1000)),\n" +
                " watermark for row_time as row_time\n" +
                ")" + KafkaUtil.getKafkaSourceConnOption("dwd_trade_order_detail", "dwd_trade_pay_detail_suc"));

        // TODO 4. 从 Kafka 读取业务数据，封装为 Flink SQL 表
        tableEnv.executeSql("create table topic_db(" +
                " `database` String,\n" +
                " `table` String,\n" +
                " `type` String,\n" +
                " `data` map<String, String>,\n" +
                " `old` map<String, String>,\n" +
                " `pt` as PROCTIME(),\n" +
                " `ts` string,\n" +
                " row_time as TO_TIMESTAMP(FROM_UNIXTIME(cast(ts as bigint)*1000)),\n" +
                " watermark for row_time as row_time" +
                ")" + KafkaUtil.getKafkaSourceConnOption("topic_db", "dwd_trade_pay_detail_suc"));

        // TODO 5. 筛选支付成功数据
        Table paymentInfo = tableEnv.sqlQuery("select\n" +
                " data['order_id'] order_id,\n" +
                " data['payment_type'] payment_type,\n" +
                " data['callback_time'] callback_time,\n" +
                " `pt`,\n" +
                "row_time,\n" +
                " ts\n" +
                " from topic_db\n" +
                " where `table` = 'payment_info'\n" +
                " and data['callback_time'] is not null"
        );
        tableEnv.createTemporaryView("payment_info", paymentInfo);


        // TODO 6. 关联
        Table resultTable = tableEnv.sqlQuery("" +
                "select \n" +
                "  od.id order_detail_id,\n" +
                "  od.course_id,\n" +
                "  od.course_name,\n" +
                "  od.order_id,\n" +
                "  od.user_id,\n" +
                "  od.origin_amount,\n" +
                "  od.coupon_reduce,\n" +
                "  od.final_amount,\n" +
                "  od.session_id,\n" +
                "  od.create_time,\n" +
                "  od.update_time,\n" +
                "  od.privince_id,\n" +
                "  od.order_status,\n" +
                "  od.sc,\n" +
                "  pi.payment_type,\n" +
                "  pi.ts,\n" +
                "  pi.callback_time\n" +
                "from dwd_trade_order_detail od\n" +
                "join payment_info pi\n" +
                "on od.order_id=pi.order_id\n" +
                "and od.row_time >= pi.row_time - INTERVAL '15' MINUTE \n" +
                "and od.row_time <= pi.row_time + INTERVAL '5' SECOND");
        tableEnv.createTemporaryView("result_table", resultTable);




        // TODO 8. 创建 Kafka dwd_trade_pay_detail_suc 表
        tableEnv.executeSql("create table dwd_trade_pay_detail_suc(\n" +
                " order_detail_id STRING,\n" +
                " course_id STRING,\n" +
                " course_name STRING,\n" +
                " order_id STRING,\n" +
                " user_id STRING,\n" +
                " origin_amount STRING,\n" +
                " coupon_reduce STRING,\n" +
                " final_amount STRING,\n" +
                " session_id STRING,\n" +
                " create_time STRING,\n" +
                " update_time STRING,\n" +
                " privince_id STRING,\n" +
                " order_status STRING,\n" +
                " sc STRING,\n" +
                " payment_type STRING,\n" +
                " ts STRING,\n" +
                " callback_time STRING," +
                "primary key(order_detail_id) not enforced\n" +
                ")" + KafkaUtil.getKafkaUpsertSinkConnOption("dwd_trade_pay_detail_suc"));

        // TODO 9. 将关联结果写入 Upsert-Kafka 表
        tableEnv.executeSql("" +
                "insert into dwd_trade_pay_detail_suc select * from result_table");
    }
}
